/**
 * 
 */
package com.dianwoba.core.redis.scalable;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor.CallerRunsPolicy;
import java.util.concurrent.TimeUnit;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.dianwoba.core.redis.connection.ShardedJedisSentinelPool;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.ScanResult;
import redis.clients.jedis.ShardedJedis;
import redis.clients.jedis.Tuple;
import redis.clients.util.SafeEncoder;

/**
 * 分片扩容管理器
 * 
 * @author ren
 *
 */
public class ScalableManager {
	
	private Logger logger = LoggerFactory.getLogger(getClass());
	// 老redis集群各分片连接地址
	private List<JedisPool> srcJedisPool;
	// 新分片jedis数据源连接池
	private  ShardedJedisSentinelPool destJedisPool;
	
	private static ExecutorService  excutor;

	public ScalableManager(List<JedisPool> srcJedisPool, ShardedJedisSentinelPool destJedisPool) {
		this.srcJedisPool = srcJedisPool;
		this.destJedisPool = destJedisPool;
		excutor = new ThreadPoolExecutor(20, 300, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(30000), 
				Executors.defaultThreadFactory(), new CallerRunsPolicy());
	}
	
	public void doMove() {
		if (CollectionUtils.isEmpty(srcJedisPool)) {
			return ;
		}
		
		for (JedisPool jedisPool : srcJedisPool) {
			
			ScanResult<byte[]> scanResult = scan(jedisPool, SafeEncoder.encode("0"));
			
			while (!SafeEncoder.encode("0").equals(scanResult.getCursorAsBytes())) {
				
				DataMover DataMover = new DataMover(jedisPool, scanResult);
				excutor.submit(DataMover);
				scanResult = scan(jedisPool, scanResult.getCursorAsBytes());
			}
		}
	}
	
	class DataMover implements Runnable {
		
		private JedisPool jedisPool;
		
		private ScanResult<byte[]> scanResult;
		
		public DataMover(JedisPool jedisPool, ScanResult<byte[]> scanResult) {
			this.jedisPool = jedisPool;
			this.scanResult = scanResult;
		}

		@Override
		public void run() {
			
			for (byte[] bts : scanResult.getResult()) {
					if ("string".equals(type(jedisPool, bts))) {
						moveValue(jedisPool, bts);
					    continue ;
					} 
					if ("list".equals(type(jedisPool, bts))) {
						moveList(jedisPool, bts);
						continue ;
					}
					if ("set".equals(type(jedisPool, bts))) {
						moveSet(jedisPool, bts);
						continue ;
					}
					if ("zset".equals(type(jedisPool, bts))) {
						moveSortSet(jedisPool, bts);
						continue ;
					}
					if ("hash".equals(type(jedisPool, bts))) {
						moveHash(jedisPool, bts);
						continue ;
					}	
			}
			//logger.info("batch move success, cursor=" + scanResult.getStringCursor());
			//System.out.println("batch move success, cursor=" + scanResult.getStringCursor());
		}
	}
	
	private ScanResult<byte[]> scan(JedisPool jedisPool, byte[] cursor) {
		Jedis jedis = null;
		try {
			jedis = jedisPool.getResource();
			ScanResult<byte[]> scanResult = jedis.scan(cursor);
			jedisPool.returnResourceObject(jedis);
			return scanResult;
		} catch (Exception e) {
			if (null != jedis) {
				jedisPool.returnResourceObject(jedis);
			}
			
			logger.error("move fail, cursor={}, e={}", SafeEncoder.encode(cursor), e.getMessage());
			System.out.println("move fail, cursor=" + SafeEncoder.encode(cursor)  + ", e=" + e.getMessage());
		}
		return null;
	}
	
	private String type(JedisPool jedisPool, byte[] key) {
		Jedis jedis = null;
		try {
			jedis = jedisPool.getResource();
			String type = jedis.type(key);
			jedisPool.returnResourceObject(jedis);
			return type;
		} catch (Exception e) {
			if (null != jedis) {
				jedisPool.returnResourceObject(jedis);
			}
			
			logger.error("move fail, key={}, e={}", key, e.getMessage());
			System.out.println("move fail, key=" + key  + ", e=" + e.getMessage());
		}
		return null;
	}
	
	private void moveValue(JedisPool jedisPool, byte[] key) {
		Jedis srcJedis = null;
		ShardedJedis destJedis = null;
		try {
			srcJedis = jedisPool.getResource();
			byte[] srcValue = srcJedis.get(key);
			jedisPool.returnResourceObject(srcJedis);
			
			destJedis = destJedisPool.getResource();
			destJedis.set(key, srcValue);
			destJedisPool.returnResource(destJedis);
		} catch (Exception e) {
			if (null != srcJedis) {
				jedisPool.returnResourceObject(srcJedis);
			}
			
			if (null != destJedis) {
				destJedisPool.returnResource(destJedis);
			}
			
			logger.error("move fail, key={}, e={}", key, e.getMessage());
			System.out.println("move fail, key=" + key  + ", e=" + e.getMessage());
		} 
	}
	
	private void moveList(JedisPool jedisPool, byte[] key) {
		Jedis srcJedis = null;
		ShardedJedis destJedis = null;
		try {
			srcJedis = jedisPool.getResource();
			List<byte[]> oldElements = srcJedis.lrange(key, 0, -1);
			jedisPool.returnResourceObject(srcJedis);
			if(CollectionUtils.isEmpty(oldElements)) {
				return ;
			} 
			destJedis = destJedisPool.getResource();
			destJedis.rpush(key, oldElements.toArray(new byte[][]{}));
			destJedisPool.returnResource(destJedis);
		} catch (Exception e) {
			if (null != srcJedis) {
				jedisPool.returnResourceObject(srcJedis);
			}
			
			if (null != destJedis) {
				destJedisPool.returnResource(destJedis);
			}
			
			logger.error("move fail, key={}, e={}", key, e.getMessage());
			System.out.println("move fail, key=" + key  + ", e=" + e.getMessage());
		} 
	}
	
	private void moveSet(JedisPool jedisPool, byte[] key) {
		Jedis srcJedis = null;
		ShardedJedis destJedis = null;
		try {
			srcJedis = jedisPool.getResource();
			Set<byte[]> memebers = srcJedis.smembers(key);
			jedisPool.returnResourceObject(srcJedis);
			
			if(CollectionUtils.isEmpty(memebers)) {
				return ;
			} 
			destJedis = destJedisPool.getResource();
			destJedis.sadd(key, memebers.toArray(new byte[][]{}));
			destJedisPool.returnResource(destJedis);
		} catch (Exception e) {
			if (null != srcJedis) {
				jedisPool.returnResourceObject(srcJedis);
			}
			
			if (null != destJedis) {
				destJedisPool.returnResource(destJedis);
			}
			
			logger.error("move fail, key={}, e={}", key, e.getMessage());
			System.out.println("move fail, key=" + key  + ", e=" + e.getMessage());
		} 
	}
	
	private void moveSortSet(JedisPool jedisPool, byte[] key) {
		Jedis srcJedis = null;
		ShardedJedis destJedis = null;
		try {
        	srcJedis = jedisPool.getResource();
    		Set<Tuple> zmembers = srcJedis.zrangeWithScores(key, 0, -1);
    		jedisPool.returnResourceObject(srcJedis);
    		
    		if(CollectionUtils.isEmpty(zmembers)) {
    			return ;
    		}
    		Map<byte[], Double> destMembers = new HashMap<byte[], Double>();
    		for(Tuple tuple : zmembers) {
    			destMembers.put(tuple.getBinaryElement(), tuple.getScore());
    		}
    		destJedis = destJedisPool.getResource();
    		destJedis.zadd(key, destMembers);
    		destJedisPool.returnResource(destJedis);
		} catch (Exception e) {
			if (null != srcJedis) {
				jedisPool.returnResourceObject(srcJedis);
			}
			
			if (null != destJedis) {
				destJedisPool.returnResource(destJedis);
			}
			
			logger.error("move fail, key={}, e={}", key, e.getMessage());
			System.out.println("move fail, key=" + key  + ", e=" + e.getMessage());
		} 
	}
	
	private void moveHash(JedisPool jedisPool, byte[] key) {
		Jedis srcJedis = null;
		ShardedJedis destJedis = null;
        try {
        	srcJedis = jedisPool.getResource();
    		Map<byte[], byte[]> hashEntry = srcJedis.hgetAll(key);
    		jedisPool.returnResourceObject(srcJedis);
    		
    		if(MapUtils.isEmpty(hashEntry)) {
    			return ;
    		}
    		destJedis = destJedisPool.getResource();
    		destJedis.hmset(key, hashEntry);
    		destJedisPool.returnResource(destJedis);
		} catch (Exception e) {
			if (null != srcJedis) {
				jedisPool.returnResourceObject(srcJedis);
			}
			
			if (null != destJedis) {
				destJedisPool.returnResource(destJedis);
			}
			
			logger.error("move fail, key={}, e={}", key, e.getMessage());
			System.out.println("move fail, key=" + key  + ", e=" + e.getMessage());
		} 
	}
	
}
